17. 格式内容与逻辑错误处理

本章概要

  • 学习材料:含混合日期、异常数值和逻辑约束的业务表。
  • 本章任务:运行 lst-date-format-errors,依次做解析、范围检查和跨字段逻辑检查。
  • 完成后你将得到:错误类型计数、清洗前后行数和质量状态表。
  • 自我检查:保留原始列并核对每类移除数;日期不可解析或逻辑冲突未处置时先检查原因。
  • 拓展练习:把质量规则拓展应用到另一中国订单表。

数据错误:比缺失值更危险的隐患

在金融数据分析中,除了缺失值外,错误数据同样普遍且危害更大。

错误数据主要分为两类:

  • 格式错误:数据存储形式不正确(如日期、数值格式混乱)
  • 逻辑错误:数据值违反业务规则(如负价格、异常涨跌幅)

逻辑错误直接破坏了数据的准确性有效性,可能导致严重的投资决策失误。

格式错误的常见类型

错误类型 示例 影响
日期格式混乱 ‘2024-01-01’ 与 ‘2024/01/01’ 混用 时间排序失败
数值类型错误 价格存储为字符串 ‘100.5’ 无法进行数学运算
千分位符号 ‘12,000’ 含逗号 转换数值报错
单位不一致 ‘元’ 与 ‘万元’ 混用 计算结果偏差巨大

逻辑错误的常见类型

错误类型 示例 检测方法
负价格 股票价格为 -50 价格 ≤ 0
异常高价 PE 超过 1000 阈值检测
成交量异常 成交量 > 流通股本 交叉验证
违反约束 最高价 < 最低价 逻辑规则检验

数据质量的六个维度

数据质量管理理论将数据质量分为六个维度(Wang & Strong, 1996):

  • 准确性(Accuracy):数据正确反映现实
  • 完整性(Completeness):无缺失值
  • 一致性(Consistency):数据间无矛盾
  • 时效性(Timeliness):数据及时更新
  • 有效性(Validity):符合业务规则
  • 唯一性(Uniqueness):无重复记录

运行前预测|平台任务解答代码

  • 输入预测:运行前先写出 dfdf1 的业务含义、数据类型或取值范围,并判断哪一个输入最可能改变结果。
  • 结果预测:不展开答案,先预测将得到df1 的结果;同时写出方向、数量级或表格/图形结构。
  • 完成要求:能独立说明本任务从输入到“平台任务解答代码”结果的关键步骤,原样录入平台代码并得到可核对的运行结果。

⭐ 平台任务解答代码

展开完整代码(投影默认折叠)
# ⚠️ 平台原始代码 - 请原样输入至教学平台(注释除外),平台才会判定答案正确
import pandas as pd  # 导入Pandas数据分析库
import numpy as np  # 导入NumPy数值计算库
# 从Excel文件读取数据存入df
df=pd.read_excel('https://huoran.oss-cn-shenzhen.aliyuncs.com/20230313/xls/1635115220651761664.xls',sheet_name='Sheet1')
df1=df.drop([247])  # 删除指定行或列
df1.dropna(how='all',inplace=True)  # 删除全部为空值的行
# 修改日期并转换数据类型
df1.iloc[201,0]='2022-11-04'
df1['日期']=pd.to_datetime(df1['日期'])  # 转换为日期时间格式
# 修改指定值
df1.loc[138,'科创50']=np.nan
# 填充缺失值
df1.fillna(method='ffill',inplace=True)
print(df1)  # 输出数据框数据

任务复盘|平台任务解答代码

运行后核对:核对 dfdf1 是否按预测参与运算,实际输出是否与预测一致;若不一致,先检查类型、单位、索引/字段和运算顺序。

拓展练习:只改变一个关键输入或业务场景,先预测输出如何变化,再运行验证并解释变化原因。

日期格式统一:问题与挑战

金融数据常来源于多个系统,日期格式不统一:

格式名称 示例 说明
ISO标准 ‘2024-01-01’ 最推荐的格式
斜杠分隔 ‘2024/01/02’ 常见于Excel
欧洲格式 ‘01-03-2024’ DD-MM-YYYY
点号分隔 ‘2024.01.04’ 不常见
紧凑格式 ‘20240105’ 无分隔符

核心工具:pd.to_datetime() 函数

日期格式统一:代码实现

Listing 1: 处理日期格式不一致问题
展开日期格式样例代码
import pandas as pd
import numpy as np

# 创建含多种日期格式的数据
data = {
    'Date': ['2024-01-01',   # ISO标准
             '2024/01/02',   # 斜杠分隔
             '01-03-2024',   # 欧洲格式
             '2024.01.04',   # 点号分隔
             '20240105'],    # 紧凑格式
    'Price': [100.5, 101.2, 99.8, 102.3, 101.5],
    'Volume': [10000, 15000, 12000, 18000, 14000]
}
df_dates = pd.DataFrame(data)

print(f'日期样例: {df_dates["Date"].tolist()};Date类型: {df_dates["Date"].dtype}')  # 输出日期格式与类型摘要
日期样例: ['2024-01-01', '2024/01/02', '01-03-2024', '2024.01.04', '20240105'];Date类型: object

方法1:pd.to_datetime 自动解析

Listing 2: pd.to_datetime自动解析日期
展开自动解析代码
# errors='coerce': 无法解析的格式转为 NaT(Not a Time)
df_dates['Date_Parsed'] = pd.to_datetime(df_dates['Date'], errors='coerce')
print(f'自动解析: {df_dates["Date_Parsed"].notna().sum()}/{len(df_dates)} 成功;失败值: {df_dates.loc[df_dates["Date_Parsed"].isna(), "Date"].tolist()}')  # 输出解析成功数与失败值
自动解析: 1/5 成功;失败值: ['2024/01/02', '01-03-2024', '2024.01.04', '20240105']
  • pandas 2 默认按首个样式严格推断,其他不匹配格式(含歧义值)转为 NaT
  • 混合格式应显式使用 format='mixed';特殊格式应指定 format

方法2 & 3:混合格式与特殊格式

Listing 3: 混合格式与特殊格式解析
# 方法2: format='mixed' 允许混合多种格式
df_dates['Date_Format2'] = pd.to_datetime(
    df_dates['Date'], format='mixed', errors='coerce'
)
print('使用format="mixed"参数:')
print(df_dates[['Date', 'Date_Format2']].head(3))

# 方法3: 对特殊格式明确指定
special_dates = pd.to_datetime('20240105', format='%Y%m%d')
print(f'\n特殊格式解析: {special_dates}')
使用format="mixed"参数:
         Date Date_Format2
0  2024-01-01   2024-01-01
1  2024/01/02   2024-01-02
2  01-03-2024   2024-01-03

特殊格式解析: 2024-01-05 00:00:00

errors 参数的三种模式

参数值 行为 适用场景
'raise'(默认) 遇到无法解析的值抛出异常 数据质量有保证时
'coerce' 无法解析转为 NaT 数据质量不确定时(推荐)
'ignore' 保留原值,不转换 需要保留原始数据时

NaT 处理方法

  • 检测:pd.isna()pd.isnull()
  • 处理:删除或手动修正

数值类型转换:问题场景

从 Excel 或 CSV 导入的数据,数值列常被识别为文本:

  • 价格列含错误值 'error' → 直接 astype(float) 会报错
  • PE 列含 'N/A' → 无法转换为数值
  • 成交量含千分位逗号 '12,000' → 转换失败

核心解决方案pd.to_numeric() + errors='coerce'

数值类型转换:代码实现

Listing 4: 将文本型数值转换为数值类型
展开文本型数值样例代码
import pandas as pd

data = {
    'Code': ['600001', '600002', '600003', '600004'],
    'Price': ['100.5', '105.2', 'error', '98.7'],
    'PE': ['15.3', 'N/A', '18.5', '22.1'],
    'Volume': ['10000', '15000', '12,000', '18,000']
}
df_numeric = pd.DataFrame(data)

print(f'文本型数值列: {df_numeric.dtypes[df_numeric.dtypes == "object"].index.tolist()};待处理值: error、N/A、千分位逗号')  # 输出需转换的列与典型错误
文本型数值列: ['Code', 'Price', 'PE', 'Volume'];待处理值: error、N/A、千分位逗号

pd.to_numeric vs astype

Listing 5: pd.to_numeric安全转换
展开安全数值转换代码
# 错误方式: astype(float) 遇到 'error' 会崩溃
# df_numeric['Price'] = df_numeric['Price'].astype(float)  # ValueError!

# 正确方式: pd.to_numeric + errors='coerce'
df_numeric['Price_Num'] = pd.to_numeric(df_numeric['Price'], errors='coerce')
df_numeric['PE_Num'] = pd.to_numeric(df_numeric['PE'], errors='coerce')

print(f'安全转换完成: Price失败 {df_numeric["Price_Num"].isna().sum()} 条,PE失败 {df_numeric["PE_Num"].isna().sum()} 条;结果类型均为 float64')  # 输出转换失败数与结果类型
安全转换完成: Price失败 1 条,PE失败 1 条;结果类型均为 float64

处理千分位逗号

Listing 6: 处理千分位逗号
展开千分位清洗代码
# 先移除逗号,再转换为数值
df_numeric['Volume_Cleaned'] = df_numeric['Volume'].str.replace(',', '')
df_numeric['Volume_Num'] = pd.to_numeric(
    df_numeric['Volume_Cleaned'], errors='coerce'
)

print(f'千分位清洗: {df_numeric["Volume"].tolist()}{df_numeric["Volume_Num"].astype(int).tolist()}')  # 对照输出清洗前后成交量
千分位清洗: ['10000', '15000', '12,000', '18,000'] → [10000, 15000, 12000, 18000]

A股数据常见格式问题

  • 价格含货币符号(¥100.5)
  • 成交量含 ‘万’、‘亿’ 单位
  • 百分比存储为 ‘12.5%’ 文本

综合格式清洗:创建脏数据

Listing 7: 创建含多种格式错误的模拟数据
展开综合脏数据构造代码
import pandas as pd
import numpy as np

data_dirty = {
    'Date': ['2024-01-01', '2024/01/02', '20240103',
             '2024-01-04', 'invalid'],
    'Code': ['600001.SH', '600002.SH', '000001.SZ',
             '600004.SH', '600005.SH'],
    'Open': ['¥10.5', '11.2', 'error', '12.5', '10.8'],
    'Close': ['10.8', '¥11.5', '12.3', '12.7', '10.5%'],
    'Volume': ['10000', '15,000', '12,000股', '18000', '20万'],
    'PE': ['15.3', 'N/A', '18.5', '>100', '22.1']
}
df_dirty = pd.DataFrame(data_dirty)

print(f'待清洗数据: {df_dirty.shape[0]} 行×{df_dirty.shape[1]} 列;错误类型含日期、货币符号、千分位、单位与比率')  # 输出数据规模与错误类型摘要
待清洗数据: 5 行×6 列;错误类型含日期、货币符号、千分位、单位与比率

综合清洗流程:步骤1-2

Listing 8: 日期统一与价格列清洗
展开日期与价格清洗代码
df_clean = df_dirty.copy()

# 步骤1:日期格式统一
df_clean['Date'] = pd.to_datetime(df_clean['Date'], errors='coerce')

# 步骤2:移除货币符号并转换为数值
for col in ['Open', 'Close']:
    df_clean[col] = (df_clean[col].astype(str)
                     .str.replace('¥', '')
                     .str.replace('$', '')
                     .str.replace('%', ''))
    df_clean[col] = pd.to_numeric(df_clean[col], errors='coerce')

print(f'步骤1–2: 日期解析失败 {df_clean["Date"].isna().sum()} 条;Open失败 {df_clean["Open"].isna().sum()} 条;Close失败 {df_clean["Close"].isna().sum()} 条')  # 输出日期与价格清洗摘要
步骤1–2: 日期解析失败 3 条;Open失败 1 条;Close失败 0 条

综合清洗流程:步骤3-4

Listing 9: 成交量与PE比率清洗
展开成交量与PE清洗代码
# 步骤3:成交量清洗(自定义函数处理多种单位)
def clean_volume(vol_str):
    '''清洗成交量数据,处理万/亿/股等单位'''
    if pd.isna(vol_str):
        return np.nan
    vol_str = str(vol_str).strip()
    if '万' in vol_str:
        return float(vol_str.replace('万', '').replace(',', '')) * 10000
    elif '亿' in vol_str:
        return float(vol_str.replace('亿', '').replace(',', '')) * 100000000
    elif '股' in vol_str:
        return float(vol_str.replace('股', '').replace(',', ''))
    else:
        return float(vol_str.replace(',', ''))

df_clean['Volume_Num'] = df_clean['Volume'].apply(clean_volume)

# 步骤4:PE比率清洗
df_clean['PE'] = df_clean['PE'].astype(str).str.replace('>', '')
df_clean['PE'] = pd.to_numeric(df_clean['PE'], errors='coerce')
print(f'步骤3–4: 成交量成功 {df_clean["Volume_Num"].notna().sum()}/{len(df_clean)} 条;PE失败 {df_clean["PE"].isna().sum()} 条')  # 输出成交量与PE清洗摘要
步骤3–4: 成交量成功 5/5 条;PE失败 1 条

清洗后验证

Listing 10: 清洗后数据质量报告
展开数据质量验证代码
missing_summary = df_clean.isnull().sum()  # 统计各列缺失值
print(f'清洗验证: 总行数 {len(df_clean)},完整行数 {df_clean.dropna().shape[0]},缺失合计 {missing_summary.sum()};缺失列 {missing_summary[missing_summary > 0].to_dict()}')  # 输出数据质量摘要
清洗验证: 总行数 5,完整行数 2,缺失合计 5;缺失列 {'Date': 3, 'Open': 1, 'PE': 1}

逻辑错误检测:价格异常

基于业务规则检测价格数据中不可能的值:

  • 规则1:价格不能为负或零(\(Price_t > 0\)
  • 规则2:价格不能超过前一日的 300%(异常涨幅)
  • 规则3:价格不能低于前一日的 30%(异常跌幅)

核心思路:将业务规则转化为布尔条件,筛选违规记录

价格异常检测:代码实现

Listing 11: 检测和处理价格异常值
展开完整参考实现
import pandas as pd
import numpy as np

np.random.seed(42)
dates = pd.date_range('2024-01-01', periods=20)
normal_prices = 100 + np.random.randn(20).cumsum() * 0.5

# 人为添加异常值
prices = normal_prices.copy()
prices[5] = -50      # 异常1:负价格
prices[10] = 5000    # 异常2:异常高价
prices[15] = 0.01    # 异常3:接近零

df_prices = pd.DataFrame({'Date': dates, 'Price': prices, 'Volume': np.random.randint(10000, 100000, 20)})  # 构造价格与成交量样例表

# 异常检测规则
rule1 = df_prices['Price'] <= 0
df_prices['Prev_Price'] = df_prices['Price'].shift(1)
rule2 = df_prices['Price'] > df_prices['Prev_Price'] * 3
rule3 = df_prices['Price'] < df_prices['Prev_Price'] * 0.3
anomaly_mask = rule1 | rule2 | rule3

print(f'异常记录数: {anomaly_mask.sum()}')
anomaly_records = df_prices[anomaly_mask].copy()
anomaly_records['Reason'] = ''
anomaly_records.loc[rule1, 'Reason'] = '非正价格'
anomaly_records.loc[rule2, 'Reason'] = '异常涨幅'
anomaly_records.loc[rule3, 'Reason'] = '异常跌幅'
print('\n异常记录详情:')
print(anomaly_records[['Date', 'Price', 'Reason']])
异常记录数: 6

异常记录详情:
         Date        Price Reason
5  2024-01-06   -50.000000   异常跌幅
6  2024-01-07   101.820045   异常涨幅
10 2024-01-11  5000.000000   异常涨幅
11 2024-01-12   101.775732   异常跌幅
15 2024-01-16     0.010000   异常跌幅
16 2024-01-17    99.290055   异常涨幅

异常处理的三种策略

Listing 12: 三种异常处理策略对比
展开三种异常处理策略
# 策略1:删除异常记录
df_clean_drop = df_prices[~anomaly_mask].copy()

# 策略2:用前值填充
df_clean_ffill = df_prices.copy()
df_clean_ffill.loc[anomaly_mask, 'Price'] = np.nan
df_clean_ffill['Price'] = df_clean_ffill['Price'].ffill()

# 策略3:标记但不删除(保留用于人工审核)
df_marked = df_prices.copy()
df_marked['Is_Anomaly'] = anomaly_mask
print(f'策略对比: 原始 {len(df_prices)} 条;删除后 {len(df_clean_drop)} 条;前值填充 {anomaly_mask.sum()} 条;人工审核标记 {df_marked["Is_Anomaly"].sum()} 条')  # 输出三种策略的可比结果
策略对比: 原始 20 条;删除后 14 条;前值填充 6 条;人工审核标记 6 条

异常检测的统计学方法

Z-Score 方法\(Z = \frac{x - \mu}{\sigma}\),通常认为 \(|Z| > 3\) 为异常值

IQR 方法:异常值 < \(Q_1 - 1.5 \times IQR\) 或 > \(Q_3 + 1.5 \times IQR\),其中 \(IQR = Q_3 - Q_1\)(四分位距)

成交额异常检测(课堂数据)

课堂数据(非市场观测):价格单位为元/股,成交量单位为股,成交额单位为元;基线严格满足“价格 × 成交量 = 成交额”,本活动注入并检测的是成交额缺陷。

Listing 13: 单位一致的成交额基线与缺陷注入
展开单位基线与缺陷注入代码
import pandas as pd  # 构造带稳定行标识的课堂数据表
import numpy as np  # 生成确定性数值序列并支持后续核对
observation_count = 50  # 固定课堂样例行数,便于比较每次运行结果
row_numbers = np.arange(1, observation_count + 1)  # 生成稳定的行序号供缺陷追踪
df_vol = pd.DataFrame({'Row_ID': [f'OBS-{row_number:03d}' for row_number in row_numbers], 'Date': pd.date_range('2024-01-01', periods=observation_count), 'Price_Yuan_Per_Share': 100 + (row_numbers % 10) * 0.2, 'Volume_Shares': 1_000_000 + (row_numbers % 7) * 100_000})  # 按元/股与股构造非市场观测的课堂基线
df_vol['Turnover_Yuan'] = df_vol['Price_Yuan_Per_Share'] * df_vol['Volume_Shares']  # 以统一单位恒等式生成基线成交额
df_vol['Injected_Defect'] = '无'  # 默认标记基线行为无缺陷
defect_plan = {'OBS-011': ('成交额放大100倍', 100.0), 'OBS-026': ('成交额缩小100倍', 0.01), 'OBS-041': ('成交额放大25%', 1.25)}  # 声明稳定行、缺陷类别与注入倍数
for row_id, (defect_category, turnover_multiplier) in defect_plan.items():  # 按固定清单逐条注入已知缺陷
    target_row_mask = df_vol['Row_ID'].eq(row_id)  # 通过稳定行标识定位注入目标
    df_vol.loc[target_row_mask, 'Turnover_Yuan'] *= turnover_multiplier  # 仅扰动成交额以保留价格与成交量基准
    df_vol.loc[target_row_mask, 'Injected_Defect'] = defect_category  # 写入可核对的缺陷类别
injected_defect_text = ';'.join(f'{row_id}={category}' for row_id, (category, _) in defect_plan.items())  # 组装紧凑的缺陷依据文本
print(f'课堂基线: {observation_count}行;Price=元/股;Volume=股;Turnover=元')  # 显示数据性质与字段单位
print(f'确定性注入: {injected_defect_text}')  # 显示预期缺陷的行标识与类别
课堂基线: 50行;Price=元/股;Volume=股;Turnover=元
确定性注入: OBS-011=成交额放大100倍;OBS-026=成交额缩小100倍;OBS-041=成交额放大25%

成交额交叉验证

Listing 14: 成交额与成交量的交叉验证
展开成交额交叉验证代码
df_vol['Expected_Turnover_Yuan'] = df_vol['Price_Yuan_Per_Share'] * df_vol['Volume_Shares']  # 使用元/股与股重新计算同单位成交额
df_vol['Reconciliation_Ratio'] = df_vol['Turnover_Yuan'] / df_vol['Expected_Turnover_Yuan']  # 计算记录值相对恒等式基准的比例
expected_defect_mask = df_vol['Injected_Defect'].ne('无')  # 从注入清单生成预期缺陷真值集
detected_defect_mask = ~np.isclose(df_vol['Reconciliation_Ratio'], 1.0, rtol=0.02, atol=0.0)  # 以2%相对容差检测核对差异:容差须覆盖行情四舍五入与最小报价单位误差,小于它会产生误报,大于它可能漏检轻度放大缺陷
expected_defect_count = int(expected_defect_mask.sum())  # 统计预期缺陷数作为检查分母
detected_defect_count = int(detected_defect_mask.sum())  # 统计检测器实际标记数
true_positive_count = int((expected_defect_mask & detected_defect_mask).sum())  # 计算预期与检测集合的交集
false_positive_count = int((~expected_defect_mask & detected_defect_mask).sum())  # 计算清洁行被误标的数量
false_negative_count = int((expected_defect_mask & ~detected_defect_mask).sum())  # 计算已知缺陷被漏标的数量
clean_ratio_median = df_vol.loc[~expected_defect_mask, 'Reconciliation_Ratio'].median()  # 用清洁行中位数验证基线接近1
is_detector_valid = true_positive_count == expected_defect_count and false_positive_count == 0 and false_negative_count == 0  # 检查检测器是否完整命中且无误报
initial_check_result = '请先修正已发现的问题' if detected_defect_count else '可以继续使用'  # 根据尚未修正的问题判断能否继续分析
detected_row_text = '、'.join(df_vol.loc[detected_defect_mask, 'Row_ID'])  # 列出实际检出的稳定行标识
print(f'核对基线: 清洁行比例中位数={clean_ratio_median:.4f};预期={expected_defect_count},检测={detected_defect_count}')  # 显示基线与预期实际计数
print(f'重叠检查: TP={true_positive_count},FP={false_positive_count},FN={false_negative_count};检测器={"通过" if is_detector_valid else "未通过"}')  # 显示交集、误报、漏报与检查状态
print(f'初始检查结果={initial_check_result};需要处理的行={detected_row_text}')  # 在修正前说明问题所在的行
df_vol.loc[detected_defect_mask, 'Turnover_Yuan'] = df_vol.loc[detected_defect_mask, 'Expected_Turnover_Yuan']  # 按恒等式重新计算已命中成交额作为教学修复
post_reconciliation_ratio = df_vol['Turnover_Yuan'] / df_vol['Expected_Turnover_Yuan']  # 修复后重新计算全部行核对比例
is_release_ready = bool(np.isclose(post_reconciliation_ratio, 1.0, rtol=0.02, atol=0.0).all())  # 检查重新核对是否全部通过
print(f'修复动作=按价格×股数重新计算成交额;再核对检查要求={"继续使用" if is_release_ready else "暂停"}')  # 输出整改动作与最终继续使用状态
核对基线: 清洁行比例中位数=1.0000;预期=3,检测=3
重叠检查: TP=3,FP=0,FN=0;检测器=通过
初始检查结果=请先修正已发现的问题;需要处理的行=OBS-011、OBS-026、OBS-041
修复动作=按价格×股数重新计算成交额;再核对检查要求=继续使用

继续使用原则:先用同单位恒等式定位缺陷;检测器检查通过不等于数据可直接后续分析,必须修复后重新核对才能继续使用。

数据一致性检验

金融数据存在内在的逻辑约束:

  • \(Price_t > 0\):价格必须为正
  • \(High_t \geq \max(Open_t, Close_t)\):最高价不低于开盘和收盘
  • \(Low_t \leq \min(Open_t, Close_t)\):最低价不高于开盘和收盘
  • \(Return_t = \frac{Price_t - Price_{t-1}}{Price_{t-1}}\):收益率定义

通过检验这些约束是否满足,可以发现数据中的逻辑矛盾。

一致性检验:代码实现

Listing 15: 数据间一致性检验
展开完整参考实现
import pandas as pd
import numpy as np

np.random.seed(42)
n_days = 10
index_values = [3000 + i * 10 + np.random.randn() * 5 for i in range(n_days)]  # 生成指数收盘价序列
returns = pd.Series(index_values).pct_change().tolist()

df_index = pd.DataFrame({'Date': pd.date_range('2024-01-01', periods=n_days), 'Index_Close': index_values, 'Index_Open': [v + np.random.uniform(-10, 10) for v in index_values], 'Daily_Return': returns, 'Volume': np.random.randint(100000000, 500000000, n_days)})  # 构造指数一致性检验样例表

# 人为添加不一致
df_index.loc[5, 'Index_Close'] = 3100
df_index.loc[3, 'Daily_Return'] = -0.5

# 一致性检验
consist1 = df_index['Index_Close'] <= 0
consist2 = (df_index['Index_Open'] / df_index['Index_Close'] > 1.2) | (df_index['Index_Open'] / df_index['Index_Close'] < 0.8)  # 检测开收盘偏离约束
df_index['Calculated_Return'] = df_index['Index_Close'].pct_change()
consist3 = (df_index['Calculated_Return'] - df_index['Daily_Return']).abs() > 0.01  # 检测收益率不一致
inconsistency_mask = consist1 | consist2 | consist3

print(f'不一致记录数: {inconsistency_mask.sum()}')
result = df_index.loc[inconsistency_mask, ['Date', 'Index_Close', 'Daily_Return', 'Calculated_Return']]  # 提取不一致记录的关键字段
print('\n不一致详情:')
print(result)
不一致记录数: 3

不一致详情:
        Date  Index_Close  Daily_Return  Calculated_Return
3 2024-01-04  3037.615149     -0.500000           0.004755
5 2024-01-06  3100.000000      0.003291           0.020130
6 2024-01-07  3067.896064      0.006254          -0.010356

实战案例:股指数据加载与诊断

Listing 16: 股指数据加载与质量评估
展开完整参考实现
import pandas as pd  # 导入表格分析工具
import numpy as np  # 导入数值计算工具
import matplotlib.pyplot as plt  # 导入绘图工具
plt.rcParams['font.sans-serif'] = ['Source Han Serif SC']  # 设置中文字体
plt.rcParams['axes.unicode_minus'] = False  # 正确显示负号
np.random.seed(42)  # 固定随机结果便于重新运行
n = 50  # 设置模拟交易日数
dates = pd.date_range('2024-01-01', periods=n)  # 生成交易日期
index_values = 3000 + np.arange(n) * 5 + np.random.randn(n) * 20  # 生成指数基准路径
df_index = pd.DataFrame({'Date': dates, 'Open': index_values + np.random.uniform(-5, 5, n), 'High': index_values + np.random.uniform(0, 10, n), 'Low': index_values - np.random.uniform(0, 10, n), 'Close': index_values, 'Volume': np.random.randint(100000000, 500000000, n), 'Amount': np.random.randint(100000000000, 500000000000, n, dtype=np.int64)})  # 构造股指样例数据
df_index.loc[5, 'Close'] = '3,250.5'  # 注入千分位格式错误
df_index.loc[10, 'Volume'] = '200000000股'  # 注入成交量单位错误
df_index.loc[15, 'Close'] = -3000  # 注入负价格逻辑错误
df_index.loc[20, 'High'] = df_index.loc[20, 'Low'] - 100  # 注入最高价低于最低价错误
df_index.loc[25, 'Low'] = 3500  # 注入最低价高于收盘价错误
df_index.loc[30, 'Volume'] = 10000000000  # 注入异常高成交量
df_index.loc[35, 'Close'] = 5000  # 注入异常高指数
print(f'股指诊断输入: {df_index.shape[0]} 行×{df_index.shape[1]} 列;已注入 7 处格式或逻辑错误用于练习')  # 输出数据规模与练习错误数
股指诊断输入: 50 行×7 列;已注入 7 处格式或逻辑错误用于练习

实战:格式错误修复

Listing 17: 修复股指数据的格式错误
展开股指格式修复代码
df_fixed = df_index.copy()  # 复制原始数据便于对照
if df_fixed['Close'].dtype == 'object':
    df_fixed['Close'] = df_fixed['Close'].astype(str).str.replace(',', '')  # 移除价格千分位逗号
numeric_cols = ['Open', 'High', 'Low', 'Close']  # 指定价格数值列
for col in numeric_cols:
    if df_fixed[col].dtype == 'object':
        df_fixed[col] = pd.to_numeric(df_fixed[col], errors='coerce')  # 将价格安全转换为数值
if df_fixed['Volume'].dtype == 'object':
    def clean_vol(vol):  # 定义成交量格式清洗函数
        if pd.isna(vol):
            return np.nan  # 保留成交量缺失值
        vol_str = str(vol).replace('股', '').replace(',', '').replace('万', '0000')  # 移除成交量符号并换算万单位
        return pd.to_numeric(vol_str, errors='coerce')  # 安全转换成交量
    df_fixed['Volume'] = df_fixed['Volume'].apply(clean_vol)  # 逐行清洗成交量
print(f'格式修复: 价格列类型 {df_fixed[numeric_cols].dtypes.astype(str).unique().tolist()};成交量类型 {df_fixed["Volume"].dtype};转换缺失 {df_fixed[numeric_cols + ["Volume"]].isna().sum().sum()} 个')  # 输出格式修复摘要
格式修复: 价格列类型 ['float64'];成交量类型 int64;转换缺失 0 个

实战:逻辑错误检测与修复

Listing 18: 检测和修复股指数据逻辑错误
展开完整参考实现
error_price_negative = df_fixed['Close'] <= 0  # 按业务规则检测非正价格
negative_count = int(error_price_negative.sum())  # 记录负价格错误数
df_fixed.loc[error_price_negative, 'Close'] = np.nan  # 将负价格标记为缺失
df_fixed['Close'] = df_fixed['Close'].ffill()  # 使用前值修复负价格
error_high_low = (df_fixed['High'] < df_fixed['Close']) | (df_fixed['Low'] > df_fixed['Close'])  # 检测高低价约束错误
high_low_count = int(error_high_low.sum())  # 记录高低价错误数
df_fixed.loc[df_fixed['High'] < df_fixed['Close'], 'High'] = df_fixed['Close']  # 修正偏低的最高价
df_fixed.loc[df_fixed['Low'] > df_fixed['Close'], 'Low'] = df_fixed['Close']  # 修正偏高的最低价
df_fixed['Prev_Close'] = df_fixed['Close'].shift(1)  # 构造前一日收盘价
df_fixed['Daily_Change'] = (df_fixed['Close'] - df_fixed['Prev_Close']) / df_fixed['Prev_Close']  # 计算单日涨跌幅
error_excess_change = df_fixed['Daily_Change'].abs() > 0.15  # 检测超过15%的异常涨跌
volume_median = df_fixed['Volume'].median()  # 计算成交量中位数基准
error_volume = (df_fixed['Volume'] > volume_median * 10) | (df_fixed['Volume'] < volume_median * 0.01)  # 检测成交量异常
errors = [('负价格', negative_count), ('高低价错误', high_low_count), ('异常涨跌', int(error_excess_change.sum())), ('异常成交量', int(error_volume.sum()))]  # 汇总四类逻辑错误
errors = [(name, count) for name, count in errors if count > 0]  # 仅保留实际出现的错误
print(f'逻辑修复: {dict(errors)};异常涨跌日期 {df_fixed.loc[error_excess_change, "Date"].dt.strftime("%Y-%m-%d").tolist()}')  # 输出错误数与异常日期
逻辑修复: {'负价格': 1, '高低价错误': 5, '异常涨跌': 2, '异常成交量': 1};异常涨跌日期 ['2024-02-05', '2024-02-06']

实战:数据质量可视化

展开完整参考实现
fig, axes = plt.subplots(2, 2, figsize=(11, 3.8))  # 创建适合幻灯片的四子图画布并控制高度
axes[0, 0].plot(df_fixed['Date'], df_fixed['Close'], color='steelblue', linewidth=1.5)  # 绘制指数收盘价走势
axes[0, 0].scatter(df_fixed.loc[error_excess_change, 'Date'], df_fixed.loc[error_excess_change, 'Close'], color='red', s=25, label='异常涨跌')  # 标注异常涨跌点
axes[0, 0].set_title('收盘价与异常涨跌')  # 设置收盘价子图标题
returns = df_fixed['Daily_Change'].dropna()  # 取出有效日收益率
axes[0, 1].hist(returns, bins=15, color='steelblue', edgecolor='white')  # 绘制日收益率分布
axes[0, 1].axvline(returns.mean(), color='red', linestyle='--', label=f'均值 {returns.mean():.3f}')  # 标注日收益率均值
axes[0, 1].set_title('日收益率分布')  # 设置收益率子图标题
axes[1, 0].bar(range(len(df_fixed)), df_fixed['Volume'], color='coral')  # 绘制成交量序列
axes[1, 0].set_title('成交量')  # 设置成交量子图标题
df_fixed['Amplitude'] = (df_fixed['High'] - df_fixed['Low']) / df_fixed['Low'] * 100  # 计算日内振幅
axes[1, 1].plot(df_fixed['Date'], df_fixed['Amplitude'], color='seagreen', linewidth=1.3)  # 绘制日内振幅走势
axes[1, 1].axhline(df_fixed['Amplitude'].mean(), color='red', linestyle='--', label='平均振幅')  # 标注平均振幅
axes[1, 1].set_title('日内振幅(%)')  # 设置振幅子图标题
axes[0, 0].tick_params(axis='x', labelbottom=False)  # 上排行情图隐藏重复日期刻度以避免拥挤
axes[1, 1].xaxis.set_major_locator(plt.matplotlib.dates.AutoDateLocator(minticks=3, maxticks=5))  # 稀疏显示日期刻度
axes[1, 1].xaxis.set_major_formatter(plt.matplotlib.dates.DateFormatter('%m-%d'))  # 缩短日期标签
axes[1, 1].tick_params(axis='x', rotation=30)  # 旋转底部日期刻度以保持分隔
for ax in axes.flat: ax.grid(True, alpha=0.25)  # 统一添加浅色网格
for ax in (axes[0, 0], axes[0, 1], axes[1, 1]): ax.legend(fontsize=16)  # 为关键子图添加图例
plt.tight_layout()  # 紧凑排列四个子图
plt.show()  # 显示数据质量图形
missing_rate = df_fixed.isnull().sum().sum() / df_fixed.size  # 计算修复后缺失率
print(f'质量摘要: 缺失率 {missing_rate:.2%};处理 {len(errors)} 类错误;有效记录 {len(df_fixed)}/{len(df_index)}')  # 输出数据质量摘要
图中展示数据质量可视化分析;读者应依据坐标、图例与注释比较主要模式。
Figure 1: 数据质量可视化分析
质量摘要: 缺失率 0.40%;处理 4 类错误;有效记录 50/50

本章总结

格式错误处理

  • 日期:pd.to_datetime(errors='coerce')
  • 数值:pd.to_numeric(errors='coerce')
  • 特殊字符:str.replace() 先清洗再转换

逻辑错误处理

  • 基于业务规则定义布尔条件
  • 交叉验证发现不一致
  • 处理策略:删除 / 前值填充 / 标记审核

随堂练习

  • 问题 1|需要准备哪些数据?:含混合日期、异常数值和逻辑约束的业务表。
  • 问题 2|需要完成哪些操作?:运行 lst-date-format-errors,依次做解析、范围检查和跨字段逻辑检查。
  • 问题 3|应得到哪些结果?:错误类型计数、清洗前后行数和质量状态表。
  • 问题 4|怎样确认结果可靠?:保留原始列并核对每类移除数;日期不可解析或逻辑冲突未处置时先检查原因。
  • 问题 5|换一个情境,怎样继续应用?:把质量规则拓展应用到另一中国订单表。
  • 作答提示:请依次写清所用数据、分析过程、所得结果、核对方法和拓展思考。课程所需数据见前言中的下载入口;教学平台固定题按页面说明完成。

教师参考解答|答案与说明 1

  • 逐项答案:下方转换演示完整的数据清洗过程的格式识别与显式转换,另须把日期解析错误和负数量逻辑错误分列统计;格式、内容和逻辑错误机制不同,静默转换会污染后续指标;拓展应用到长三角财报时画错误类型计数图,并以原始/清洗行数与标记表核对。

  • 所用数据与字段:金额字符串、日期与业务约束

  • 参考代码或推导(平台保护块之外):

教师参考解答|代码 1

展开代码(代码区可独立滚动)
import pandas as pd  # 导入表格库
import matplotlib.pyplot as plt  # 导入绘图库生成错误计数图
raw=pd.DataFrame({'amount':['1,200','bad',None],'date':['2026-01-02','bad','2026-01-04'],'quantity':[2,-1,3]})  # 创建三类混杂输入
clean_amount=pd.to_numeric(raw['amount'].str.replace(',','',regex=False),errors='coerce')  # 识别金额格式错误
clean_date=pd.to_datetime(raw['date'],errors='coerce')  # 识别日期解析错误
quality={'amount_invalid':int(clean_amount.isna().sum()),'date_invalid':int(clean_date.isna().sum()),'negative_quantity':int(raw['quantity'].lt(0).sum())}  # 分列统计格式与逻辑错误
quality_counts=pd.Series(quality,name='错误数')  # 形成可核对图表数据
axis=quality_counts.plot.bar(color=['#EC232A','#00A4E6','#007A86'],title='错误类型计数')  # 实际绘制错误计数柱状图
axis.set_ylabel('记录数')  # 标注统计单位
plt.tight_layout(); print(quality_counts)  # 渲染图并输出同源计数表

教师参考解答|答案与说明 2

  • 参考结果:同源计数表与柱状图均显示 amount_invalid=2、date_invalid=1、negative_quantity=1;错误不被填成0
  • 边界 / 局限:格式错误与逻辑错误分开
  • 常见错误:errors=ignore;无效值填0